package tom.reactor;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import tom.reactor.model.MsgEvent;

import java.time.Duration;

/**
 *
 */
class SpringWebFluxDemo03ApplicationTest01 {

    private WebClient client;

    @BeforeEach
    public void setUp() {
        this.client = WebClient.create("http://localhost:8080");
    }

    @Test
    public void testSentEvents() {
        //从0开始逐渐增加，相当于计数器，1秒钟加一个数
        Flux<MsgEvent> flux = Flux.interval(Duration.ofSeconds(3L)).map(
                n -> new MsgEvent(n, "msg" + n)
        ).take(30);

        this.client.post().uri("/event")
                .contentType(MediaType.APPLICATION_STREAM_JSON)
                .body(flux, MsgEvent.class)
                .retrieve()
                //结果为Void类型
                .bodyToMono(Void.class)
                .log()
                //阻塞直到发送完毕flux数据
                .block();

    }

    @Test
    public void testReceiveEvents() {
        this.client.get().uri("/event")
                .accept(MediaType.APPLICATION_STREAM_JSON)
                .retrieve()
                .bodyToFlux(MsgEvent.class)
                .log()
                .take(20)
                .blockLast()
        ;
    }
}